package com.alison.datastream.chapter6_timeAndWaterMark.watermark;

import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.eventtime.TimestampAssignerSupplier;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.TimestampAssigner;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.time.Duration;
import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.time.format.DateTimeFormatter;

/**
 * @Author alison
 * @Date 2024/4/5 9:28
 * @Version 1.0
 * @Description
 */
public class W5_WatermarkStreamingJob {
    private static final Logger logger = LoggerFactory.getLogger(W5_WatermarkStreamingJob.class);

    public static void main(String[] args) throws Exception {
        // 创建 执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        // 自定义数据源,每隔1000ms下发一条数据
        SourceFunction<JSONObject> dataSource = new SourceFunction<JSONObject>() {
            private volatile boolean running = true;
            @Override
            public void run(SourceContext<JSONObject> sourceContext) throws Exception {
                while (running) {
                    long timestamp = System.currentTimeMillis();
                    // 将时间戳转换为 LocalDateTime 对象
                    LocalDateTime dateTime = LocalDateTime.ofInstant(Instant.ofEpochMilli(timestamp), ZoneId.systemDefault());
                    // 定义日期时间格式
                    DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
                    // 格式化日期时间对象为指定格式的字符串
                    String format = formatter.format(dateTime);

                    JSONObject dataObj = new JSONObject();
                    int transId = 8;
                    dataObj.put("userId", "user_" + transId);
                    dataObj.put("timestamp", timestamp);
                    dataObj.put("datetime", format);
                    dataObj.put("url", "example.com/page" + transId);

                    logger.info("数据源url={},用户={},交易时间={},系统时间={}", "example.com/page" + transId, "user_" + transId, format, timestamp);
                    Thread.sleep(1000);
                    sourceContext.collect(dataObj);
                }
            }

            @Override
            public void cancel() {
                running = false;
            }
        };

        //创建水印策略处理事件发生时间
        TimestampAssignerSupplier<JSONObject> timestampAssignerSupplier = new TimestampAssignerSupplier<JSONObject>() {
            @Override
            public TimestampAssigner<JSONObject> createTimestampAssigner(Context context) {
                return (element, recordTimestamp) -> {
                    //使用自定义的事件发生时间来做水印,确保窗口统计的是按照我们的时间字段统计，提高准确度，否则默认使用消费时间
                    return element.getLong("timestamp");
                };
            }
        };

        //创建数据流
        env.addSource(dataSource).assignTimestampsAndWatermarks(
                WatermarkStrategy.<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(1))
                        .withTimestampAssigner(timestampAssignerSupplier))
                //按照url分组
                .keyBy(new KeySelector<JSONObject, Object>() {
                    @Override
                    public Object getKey(JSONObject jsonObject) throws Exception {
                        return jsonObject.getString("url");
                    }
                })
                .window(TumblingEventTimeWindows.of(Time.seconds(2)))
                .reduce(new ReduceFunction<JSONObject>() {
                    @Override
                    public JSONObject reduce(JSONObject reduceResult, JSONObject record) throws Exception {
                        logger.info("窗口统计url={},用户流水={},次数={}", reduceResult.getString("url"),
                                reduceResult.getString("userId"),
                                reduceResult.getInteger("urlNum") == null ? 1 : reduceResult.getInteger("urlNum"));
                        int urlNum = reduceResult.getInteger("urlNum") == null ? 1 : reduceResult.getInteger("urlNum");
                        reduceResult.put("urlNum", urlNum + 1);
                        return reduceResult;
                    }
                })
                .print();

        // 执行任务
        env.execute("WatermarkStreamingJob");
    }
}
